﻿using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.ServiceModel.Channels;
using System.Text;
using System.Threading;
using Kugar.Core.BaseStruct;
using Kugar.Core.ExtMethod;
using System.Net;
using System.Net.Sockets;
using Kugar.Core.Collections;
using Kugar.Core.ExtMethod;

namespace Kugar.Core.Communications.Transfer
{
    internal enum CmdID : byte
    {
        Unknow = 0x00,
        SendCmd = 0x01,
        RevACK = 0x02,


        HandShake_SYN_ConnectRequest = 0xF0,
        HandShake_SYN_ConnectAccept = 0xF1,
        HandShake_SYN_ConnectAcceptACK = 0xF2,

        KeepActive = 0xFF,


    }

    internal enum ConnectRequest
    {

    }

    internal class RUDPConnectInfo
    {
        public int BagID;
        public int ConnectID;
        public ConnectState Step;
        public DateTime LastRevACKTime;

    }

    internal enum ConnectState
    {
        None,
        SYN_SEND,
        SYN_RECV,
        ESTABLISHED
    }

    /*
     数据包格式：
                项目            长度		序号		备注
            ProtocolVer         1byte    	0			协议版本号
            CMDID				1byte		1			指令编号
            Data				nbyte					实际数据
            CrcCode				2byte					crc校验码
    
    
        三次握手数据包
    
        1.第一次握手，由client发起 A
                项目                  长度		            序号		                       备注
            ProtocolVer         1byte    	           0			            协议版本号
            CMDID				    1byte		           1			            指令编号=0xF0 HandShake_SYN_Send
            ACK					    1byte		           2			            ACK=0
            SYN					    1byte		           2			            SYN码 =1
            SeqNumber			4byte		         3-6			            随机序列号  =x
            CrcCode				2byte		         7-8			            crc校验码	
    
        2.第二次握手，接收到0xF0后，由服务器端回复：B
                项目            长度		序号		备注
            ProtocolVer         1byte    	0			协议版本号
            CMDID				1byte		1			指令编号=0xF1 HandShake_SYN_Rev
            ACK					1byte		2			ACK码 =1	
            SYN					1byte		3			SYN码 =1
            SeqNumber			4byte		4-7			随机序列号 =y
            ACK Number			4byte		8-11		接收到的序列号+1 = A.SeqNumber+1
            CrcCode				2byte		12-13		crc校验码		
    
        3.第三次握手，接收到0xF1后，检查ACK Number=A.SeqNumber+1 and ack=1，由客户端回复： C
                项目            长度		序号		备注
            ProtocolVer         1byte    	0			协议版本号
            CMDID				1byte		1			指令编号=0xF2 HandShake_SYN_Rev_ACK
            ACK					1byte		2			=1	
            SYN					1byte		3			=1
            SeqNumber			4byte		4-7			随机序列号 =y
            ACKNumber			4byte		8-11		回复接收到的序列号+1 =B.SeqNumber+1	
            CrcCode				2byte		12-13		crc校验码		

        正常数据发送 F

                项目            长度		序号					备注
            ProtocolVer         1byte    	0					协议版本号
            CMDID				1byte		1					指令编号=0x01/0x03
            BagID/SeqNumber		4byte		2-5					随机序列号 =y
            DataTotalLength		2byte		6-7					本数据包总发送字节数	
            BagIndex			1byte		8					当前数据包分片数
            Data				nbyte		9-(n+9)				实际数据
            CrcCode				2byte		(n+10)-(n+11)		crc校验码

        接收到发送数据后： G
    
                项目            长度		序号					备注
            ProtocolVer         1byte    	0					协议版本号
            CMDID				1byte		1					指令编号=0x02
            RevACKBagID			4byte		2-5					已接收到的数据包ID = F.BagID
            RevACKBagIndex		1byte		6-7					已接收到的数据包分片数= F.BagIndex
            CrcCode				2byte		8-9					crc校验码
     
     */

    public class RUDPServer
    {
        //private bool _isEnableRUDP = true;
        private Socket _socket;
        private MutileKeysDictionary<IPEndPoint, int, RUDPConnectInfo> cacheConnectionRequest;
        private MutileKeysDictionary<IPEndPoint, int, UDPTransferInternal> cacheUDPConnection;
        public ITransfer ConnectTo(IPEndPoint remotePoint, bool isEnableRUDP)
        {
            if (isEnableRUDP)
            {

            }
        }

        public bool EnabledRUDP { set; get; }

        private void OnDataReceive(IAsyncResult iar)
        {
            int readCount = 0;
            byte[] buff = null;
            EndPoint rp = null;

            try
            {
                readCount = _socket.EndReceiveFrom(iar, ref rp);
                buff = (byte[])iar.AsyncState;
            }
            catch (Exception)
            {
            }


            if (readCount < 5)
            {
                return;
            }

            if (EnabledRUDP)
            {
                var cmdID = buff[4];

                if (cmdID == 0xF0 || cmdID == 0xF1 || cmdID == 0xF2)
                {
                    handShakeTo((IPEndPoint)rp, new ArraySegment<byte>(buff, 0, readCount));
                }
                else
                {
                    var connectID = BitConverter.ToInt16(buff, 5);

                    var trans = cacheUDPConnection.TryGetValue((IPEndPoint)rp, connectID);

                    if (trans == null)
                    {
                        return;
                    }
                    else
                    {
                        trans.OnDataReceive(new ArraySegment<byte>(buff, 0, readCount), null);
                    }
                }
            }
            else
            {

            }

        }

        /// <summary>
        ///     作为服务器端，即被连接端时，接收到 SYN 信号，并回复 SYN+ACK包,,接收Rev_ACK则建立连接
        /// 
        /// </summary>
        /// <param name="remotePoint"></param>
        /// <param name="data"></param>
        private void serverHandShake(IPEndPoint remotePoint, ArraySegment<byte> data)
        {
            CmdID cmdID = CmdID.Unknow;

            int seqNumber = 0;

            var config = cacheConnectionRequest.TryGetValue(remotePoint, connectID);

            if (cmdID == CmdID.HandShake_SYN_ConnectRequest)
            {
                if (config == null)
                {
                    config = new RUDPConnectInfo() { BagID = RandomEx.Next(10, int.MaxValue) - 1, Step = ConnectState.SYN_RECV };

                    cacheConnectionRequest.Add(remotePoint, connectID, config
                                              );


                }
                else
                {
                    if (config.Step != ConnectState.SYN_RECV || config.Step != ConnectState.ESTABLISHED)
                    {
                        return;
                    }
                }


                //TODO: 回复  SYN+ACK 包 CmdID=HandShake_SYN_ConnectAccept ,BagID=Random,ACKNumber=seqNumber+1

            }
            else if (cmdID == CmdID.HandShake_SYN_ConnectAcceptACK)
            {
                var ackNumber = 0;

                if (config == null)
                {
                    return;
                }
                else
                {
                    if (config.Step == ConnectState.ESTABLISHED)  //如果已经建立连接的话，则不处理
                    {
                        return;
                    }

                    if (config.Step == ConnectState.SYN_RECV && config.BagID != ackNumber - 1)
                    {
                        int connectID = 0;
                        int remoteBufferLength = 0;

                        config.Step = ConnectState.ESTABLISHED;

                        cacheUDPConnection.Add(remotePoint, connectID, new UDPTransferInternal(_socket, remoteBufferLength, true, remotePoint));
                    }
                }


            }

            config.LastRevACKTime = DateTime.Now;
        }

        /// <summary>
        ///     作为客户端
        /// </summary>
        /// <param name="remotePoint"></param>
        /// <param name="data"></param>
        private void clientHandShakeTo(IPEndPoint remotePoint, ArraySegment<byte> data)
        {
            var bagID = BitConverter.ToInt32(data.Array, data.Offset);
            var cmdID = (CmdID)data.Array[data.Offset + 4];
            var ackNumber = 0;
            var info = cacheConnectionRequest.TryGetValue(remotePoint, bagID);

            if (cmdID == CmdID.HandShake_SYN_ConnectAccept) //0xF1
            {
                if (info == null)
                {
                    return;
                }

                if (info.BagID != ackNumber - 1)
                {
                    return;
                }

                if (info.Step == ConnectState.ESTABLISHED)
                {
                    //TODO:返回 0xF2 ,BagID=Random,ACKNumber=bagID+1,ConnectID=info.ConnectID,LocalRevBufferLength=_socket.RevBuffer
                }
                else if (info.Step == ConnectState.SYN_SEND)
                {
                    var connectID = RandomEx.Next();
                    int remoteBufferLength = 0;
                    info.Step = ConnectState.ESTABLISHED;
                    //TODO:返回 0xF2 ,BagID=Random,ACKNumber=bagID+1,ConnectID=connectID,LocalRevBufferLength=_socket.RevBuffer

                    cacheUDPConnection.Add(remotePoint, connectID,
                                           new UDPTransferInternal(_socket, remoteBufferLength, true, remotePoint));
                }




            }

        }


    }


    internal class RUDP
    {
        private Socket _socket = null;
        private int _remoteRevBufferLength = 8192;
        private IPEndPoint _remotPoint;
        private ConnectState connectState = ConnectState.ESTABLISHED;
        private int sendErrorCount = 0;
        private int _revSlideWindowSize = 5;
        private ConcurrentDictionary<int, RUDPAsyncResult> cacheCallbackList = new ConcurrentDictionary<int, RUDPAsyncResult>();
        private short _lastRevSegmentIndex = 0;
        private List<RevItemSegment> cacheRevInOrderSegment = new List<RevItemSegment>(20);

        private ConcurrentDictionary<short, RevItemSegment> cacheRevNoOrderSegment = null;
        private List<short> cacheACK = new List<short>(10);

        private ConcurrentQueue<SendItemSegment> sendingPage = new ConcurrentQueue<SendItemSegment>();

        public RUDP(Socket socket, int remoteRevBufferLength, IPEndPoint remotPoint)
        {
            _socket = socket;
            _remoteRevBufferLength = remoteRevBufferLength;
            _remotPoint = remotPoint;

            cacheRevNoOrderSegment = new ConcurrentDictionary<short, RevItemSegment>();
        }

        public IAsyncResult BeginSend(ArraySegment<byte> data, AsyncCallback callback, object state)
        {
            if (connectState != ConnectState.ESTABLISHED)
            {
                //TODO:引发连接已断开的错误
                return null;
            }

        }

        internal void ReceiveDataInternal(ArraySegment<byte> data)
        {
            if (data.HasData() || data.Count < 17)
            {
                return;
            }

            var offset = data.Offset;
            var dataArray = data.Array;

            var cmdID = (CmdID)dataArray[offset + 1];
            var bagIndex = BitConverter.ToInt16(dataArray, offset + 6);
            var isEnd = dataArray[offset + 8] != 0;

            if (cmdID == CmdID.SendCmd)
            {
                short tempLastIndex = 0;

                unchecked
                {
                    tempLastIndex = (short)(_lastRevSegmentIndex + 1);
                }


                if (bagIndex == tempLastIndex)
                {
                    unchecked
                    {
                        _lastRevSegmentIndex += 1;
                    }

                    cacheRevInOrderSegment.Add(new RevItemSegment() { Data = data, Index = bagIndex, IsEnd = isEnd });

                    if (isEnd)
                    {
                        //TODO:引发数据包接收完成操作
                    }

                    cacheACK.Add(bagIndex);

                    if (cacheACK.Count >= _revSlideWindowSize)
                    {
                        //TODO:引发ACK包
                    }
                }
                else if (bagIndex > _lastRevSegmentIndex && bagIndex < _lastRevSegmentIndex + _revSlideWindowSize)
                {
                    cacheRevNoOrderSegment.TryAdd(bagIndex, new RevItemSegment() { Data = data, Index = bagIndex, IsEnd = isEnd });

                    foreach (var itemSegment in cacheRevNoOrderSegment)
                    {
                        if (itemSegment.Key == tempLastIndex)
                        {
                            cacheRevInOrderSegment.Add(itemSegment.Value);
                        }
                    }
                }
            }
            else
            {

            }




        }



        private ArraySegment<byte> buildSendDataPack(ArraySegment<byte> data, ref int bagID, int bagIndex)
        {

        }

        private class RUDPAsyncResult : IAsyncResult
        {
            internal int Count { set; get; }

            internal Exception Error { set; get; }

            #region Implementation of IAsyncResult

            public bool IsCompleted { get; internal set; }

            public WaitHandle AsyncWaitHandle
            {
                get { throw new NotImplementedException(); }
            }

            public object AsyncState { get; internal set; }

            public bool CompletedSynchronously
            {
                get { return true; }
            }

            #endregion



        }





        internal enum RevItemHandle
        {
            None,
            Received,
            ACK
        }



        internal struct SendItemSegment : IEquatable<SendItemSegment>
        {
            public SendItemSegment(bool isSuccess, bool isLast) : this()
            {
                IsSuccess = isSuccess;
                IsLast = isLast;
            }

            public ArraySegment<byte> Data;
            public short Index;
            public bool IsLast;
            public int BagID;
            public bool IsSuccess;
            public DateTime SendTime;

            public bool Equals(SendItemSegment other)
            {
                return other.Index == Index && other.BagID == BagID && other.IsSuccess.Equals(IsSuccess);
            }

            public override bool Equals(object obj)
            {
                if (ReferenceEquals(null, obj)) return false;
                if (obj.GetType() != typeof (SendItemSegment)) return false;
                return Equals((SendItemSegment) obj);
            }

            public override int GetHashCode()
            {
                unchecked
                {
                    int result = Index.GetHashCode();
                    result = (result*397) ^ BagID;
                    result = (result*397) ^ IsSuccess.GetHashCode();
                    return result;
                }
            }

            public static bool operator ==(SendItemSegment left, SendItemSegment right)
            {
                return left.Equals(right);
            }

            public static bool operator !=(SendItemSegment left, SendItemSegment right)
            {
                return !left.Equals(right);
            }
        }

        internal class SendItem
        {
            public ArraySegment<byte> Data;
            public int BagID;
            public int CurrentIndex;

        }

        internal class RevItemSegment : IComparer
        {
            public short Index;
            public ArraySegment<byte> Data;
            public bool IsEnd;

            #region Implementation of IComparer

            public int Compare(object x, object y)
            {
                return ((RevItemSegment)x).Index - ((RevItemSegment)y).Index;
            }

            #endregion
        }

        public class SlideWindowRevQueue
        {
            private LinkedList<RevItemSegment> _link = new LinkedList<RevItemSegment>();
            private short _startIndex;
            private readonly SortedList _windowList = new SortedList(5);
            private short _windowSize ;
            private volatile short _lastEnqueueIndex = 0;
            private readonly HashSet<short> _revedIndexCache = new HashSet<short>();
            private readonly object _lockerObj = new object();
            private readonly object _lockerBuilderObj = new object();
            private Thread _buildThread = null;
            private bool isClosed = false;
            private BufferManager _bufferManager = null;

            public SlideWindowRevQueue(short startIndex,short windowSize,BufferManager bufferManager)
            {
                if (windowSize <= 0)
                {
                    throw new ArgumentOutOfRangeException("windowSize");
                }

                if (bufferManager==null)
                {
                    throw new ArgumentNullException("bufferManager");
                }

                _startIndex = startIndex;
                _windowSize = windowSize;
                _bufferManager = bufferManager;

                for (short i = _startIndex; i < _startIndex + _windowSize; i++)
                {
                    _windowList.Add(i, null);
                }

                _buildThread = new Thread(BuildDataBag);
                _buildThread.Start();
            }

            public RevItemHandle EnQueue(RevItemSegment revItem)
            {
                Monitor.Enter(_revedIndexCache);

                //_revedIndexCache 可以改用环形队列,,可设置固定大小,一般为3倍窗口大小
                if (_revedIndexCache.Contains(revItem.Index))
                {
                    Monitor.Exit(_revedIndexCache);

                    return RevItemHandle.Received;
                }

                Monitor.Enter(_lockerObj);
                if (_windowList.ContainsKey(revItem.Index))
                {
                    _windowList[revItem.Index] = revItem;

                    if (revItem.Index == _lastEnqueueIndex)
                    {
                        var tempIndex = _lastEnqueueIndex;

                        unchecked
                        {
                            _lastEnqueueIndex++;
                            tempIndex += _windowSize;
                        }

                        Monitor.Enter(_revedIndexCache);

                        _revedIndexCache.Add(revItem.Index);

                        Monitor.Exit(_revedIndexCache);

                        _windowList.Add(tempIndex, null);
                    }

                    Monitor.PulseAll(_lockerBuilderObj);

                    Monitor.Exit(_lockerObj);

                    return RevItemHandle.ACK;
                }
                else
                {
                    Monitor.Exit(_lockerObj);
                    return RevItemHandle.None;
                }

            }

            private void BuildDataBag(object state)
            {
                Monitor.Enter(_lockerObj);

                do
                {
                    var currentIndex = _windowList.Count;
                    bool needBuild = false;

                    if (currentIndex > 0)
                    {
                        short enqueueIndex = _link.Last.Value.Index;

                        do
                        {
                            unchecked
                            {
                                enqueueIndex++;
                            }

                            var item = (RevItemSegment)_windowList.GetByIndex(0);

                            if (enqueueIndex == item.Index)
                            {
                                _windowList.Remove(item.Index);

                                _link.AddLast(item);
                                if (item.IsEnd)
                                {
                                    needBuild = true;
                                }
                            }
                            else
                            {
                                break;
                            }

                            currentIndex--;

                        } while (currentIndex < 0);
                    }

                    if (needBuild)
                    {
                        LinkedListNode<RevItemSegment> currentNode = _link.First;

                        do
                        {
                            if (currentNode.Value.IsEnd)
                            {
                                var tempItem = _link.First;
                                var buffer = _bufferManager.TakeBuffer(8192);
                                var bs = new ByteStream(buffer, 0);

                                do
                                {
                                    
                                    _link.RemoveFirst();
                                    tempItem.Value = null;

                                    bs.Write(tempItem.Value.Data.Array, tempItem.Value.Data.Offset + 9, tempItem.Value.Data.Count - 9);

                                } while (tempItem != currentNode);

                                //TODO:复制List中的数据区，并拼凑成一个数组引发收到数据的事件
                                bs.Dispose();

                                OnDataBuildSuccess(new ArraySegment<byte>(buffer, 0, (int)bs.Position));

                            }

                            currentNode = currentNode.Next;



                        } while (currentNode != null);
                    }


                    if (isClosed)
                    {
                        break;
                    }

                    Monitor.Wait(_lockerBuilderObj, 2000);
                } while (!isClosed);


                Monitor.Exit(_lockerObj);
            }
        }

        public class SlideWindowSendQueueInternal
        {
            private int _dataPageSize;
            private short _windowSize = 5;
            private ConcurrentQueue<SendItemSegment> _sendQueue = null;
            private SortedList<int, SendItemSegment> _currentSendedWindow = new SortedList<int, SendItemSegment>(5);
            private volatile short _nextSendIndex = 0;
            private int _bagID = 0;
            private object _lockObj = new object();
            private bool _isClosed = false;

            public int Enqueue(byte[] data,int offset,int count)
            {

            }

            //分解数据包为数据片,并放入发送队列
            private int Splite(byte[] data, int offset, int count)
            {
                if (data == null || data.IsInEnableRange(offset, count))
                {
                    throw new ArgumentOutOfRangeException("data");
                }

                unchecked
                {
                    Interlocked.Increment(ref _bagID);
                    //_bagID++;
                }

                var bagIndex = _bagID;

                if (count <= _dataPageSize)
                {
                    _sendQueue.Enqueue(new SendItemSegment() { Data = new ArraySegment<byte>(data, offset, count), IsLast = true, Index = _nextSendIndex, BagID = bagIndex });

                    unchecked
                    {
                        _nextSendIndex++;
                    }
                }
                else
                {
                    var pageCount = (int)Math.Ceiling(((double)count / _dataPageSize));

                    for (int i = 0; i < pageCount - 1; i++)
                    {
                        _sendQueue.Enqueue(new SendItemSegment() { Data = new ArraySegment<byte>(data, i * _dataPageSize, _dataPageSize), IsLast = false, BagID = bagIndex });

                        unchecked
                        {
                            _nextSendIndex++;
                        }
                    }

                    _sendQueue.Enqueue(new SendItemSegment() { Data = new ArraySegment<byte>(data, (pageCount - 1) * _dataPageSize, count - (pageCount - 1) * _dataPageSize), IsLast = true, BagID = bagIndex });

                    unchecked
                    {
                        _nextSendIndex++;
                    }
                }

                Monitor.PulseAll(_lockerObj);

                return bagIndex;
            }
        }

        public class SlideWindowSendQueue
        {
            private int _pageSize;
            private int _dataPageSize;
            private short _windowSize = 5;
            private ConcurrentQueue<SendItemSegment> _sendQueue = null;
            private const int headCount = 16;
            private object _lockObj=new object();
            private SortedList<int, SendItemSegment> _currentSendedWindow = new SortedList<int, SendItemSegment>(5);
            private volatile short _nextSendIndex = 0;
            private int _bagID = 0;
            private bool _isClosed = false;
            private Dictionary<int,SendAsyncObj> cacheAsyncObj=new Dictionary<int, SendAsyncObj>(10);
            private Thread _putSegmentToSendListThread = null;
            private Thread _timeOutCheckerThread = null;

            public SlideWindowSendQueue()
            {
                _putSegmentToSendListThread = new Thread(PutSegmentToSendList);
                _timeOutCheckerThread=
            }

            public IAsyncResult BeginSend(byte[] data,int offset,int count,AsyncCallback callback,object asyncState)
            {
                var bagID = Enqueue(data,offset,count);

                var temp = new SendAsyncObj(bagID, count, callback, asyncState);

                cacheAsyncObj.Add(bagID,temp);

                return temp;
            }

            public int EndSend(IAsyncResult iar)
            {
                var item = (SendAsyncObj)iar;

                if (item.Callback==null)
                {
                    return -1;
                }

                cacheAsyncObj.Remove(item.DataLength);

                if (item.HasError)
                {
                    throw item.Error;
                }
                else
                {
                    return item.DataLength;
                }
            }

            //分解数据包为数据片,并放入发送队列
            private int Enqueue(byte[] data,int offset,int count)
            {
                if (data==null || data.IsInEnableRange(offset,count))
                {
                    throw new ArgumentOutOfRangeException("data");
                }

                unchecked
                {
                    Interlocked.Increment(ref _bagID);
                    //_bagID++;
                }

                var bagIndex = _bagID;

                if (count <= _dataPageSize)
                {
                    _sendQueue.Enqueue(new SendItemSegment() { Data = new ArraySegment<byte>(data, offset, count), IsLast = true, Index = _nextSendIndex, BagID = bagIndex });

                    unchecked
                    {
                        _nextSendIndex++;
                    }
                }
                else
                {
                    var pageCount =(int)Math.Ceiling(((double)count/_dataPageSize));

                    for (int i = 0; i < pageCount-1; i ++)
                    {
                        _sendQueue.Enqueue(new SendItemSegment() { Data = new ArraySegment<byte>(data, i * _dataPageSize, _dataPageSize), IsLast = false, BagID = bagIndex });
                        
                        unchecked
                        {
                            _nextSendIndex++;
                        }
                    }

                    _sendQueue.Enqueue(new SendItemSegment() { Data = new ArraySegment<byte>(data, (pageCount - 1) * _dataPageSize, count - (pageCount - 1) * _dataPageSize), IsLast = true, BagID = bagIndex });

                    unchecked
                    {
                        _nextSendIndex++;
                    }
                }

                Monitor.PulseAll(_lockerObj);

                return bagIndex;
            }

            public void EnqueueACK(short ackIndex)
            {
                Monitor.Enter(_currentSendedWindow);

                if (_currentSendedWindow.ContainsKey(ackIndex))
                {
                    var temp = _currentSendedWindow[ackIndex];

                    temp.IsSuccess = true;

                    if (temp==_currentSendedWindow.GetByIndex(0))
                    {
                        Monitor.PulseAll(_lockerObj);
                    }
                }

                Monitor.Exit(_currentSendedWindow);
            }

            private void TimeOutChecker(object state)
            {
                Monitor.Enter(_currentSendedWindow);


                var now = DateTime.Now.AddSeconds(-10);

                foreach (var segment in _currentSendedWindow)
                {
                    if (segment.Value.SendTime > now)
                    {
                        //TODO:重发该数据片
                    }

                }

            }

            private object _lockerObj=new object();
            private void PutSegmentToSendList(object state)
            {
                Monitor.Enter(_lockerObj);

                do
                {
                    var count = _currentSendedWindow.Count;

                    while (count > 0 && count <= _currentSendedWindow.Count)
                    {
                        Monitor.Enter(_currentSendedWindow);
                        var item = _currentSendedWindow.GetByIndex(0);

                        if (!item.IsSuccess)
                        {
                            break;
                        }

                        _currentSendedWindow.RemoveAt(0);

                        Monitor.Exit(_currentSendedWindow);

                        if (item.IsLast)
                        {
                            OnSendComplete(item.BagID);
                        }
                    }

                    if (_currentSendedWindow.Count <= _windowSize)
                    {
                        var newItemCount = _windowSize - _currentSendedWindow.Count;
                        for (int i = 0; i < newItemCount; i++)
                        {
                            SendItemSegment newItem;
                            if (_sendQueue.TryDequeue(out newItem))
                            {
                                _currentSendedWindow.Add(newItem.Index,newItem);
                            }
                            else
                            {
                                break;
                            }
                        }
                    }

                    Monitor.Wait(_lockerObj,2000);

                    if (_isClosed)
                    {
                        break;
                    }

                } while (!_isClosed);

                Monitor.Exit(_lockerObj);
            }

            private void OnSendComplete(int bagID)
            {
                SendAsyncObj item;

                if (cacheAsyncObj.TryGetValue(bagID, out item))
                {
                    if (item.Callback!=null)
                    {
                        item.IsCompleted = true;
                        ThreadPool.QueueUserWorkItem(AsyncSendCompleteCallback, item);
                    }
                }
            }

            private void AsyncSendCompleteCallback(object state)
            {
                var item = (SendAsyncObj) state;

                if (item.Callback==null)
                {
                    cacheAsyncObj.Remove(item.BagID);
                    return;
                }

                item.Callback(item);
            }

        }

        internal struct SendAsyncObj:IAsyncResult
        {
            public SendAsyncObj(int bagId,int dataLength, AsyncCallback callback,  object state) : this()
            {
                DataLength = dataLength;
                Callback = callback;
                BagID = bagId;
                this.state = state;
                IsCompleted = false;
            }

            public int DataLength;
            public AsyncCallback Callback;
            public int BagID;
            public object state;
            public int ConnectID;

            public Exception Error;

            public bool HasError
            {
                get { return Error != null; }
            }

            #region Implementation of IAsyncResult

            public bool IsCompleted { get; internal set; }

            public WaitHandle AsyncWaitHandle
            {
                get { return null; }
            }

            public object AsyncState
            {
                get { return state; }
            }

            public bool CompletedSynchronously
            {
                get { return false; }
            }

            #endregion
        }

        //internal class SlideWindowRevQueue
        //{
        //    private LinkedList<RevItemSegment> revLinkList = new LinkedList<RevItemSegment>();
        //    private short _revWindowsSize = 5;
        //    private bool isClosed = false;
        //    SortedList _revNoOrderCache = new SortedList(5);
        //    private BufferManager _bufferManager = null;
        //    private Thread _threadDataBuild = null;
        //    private HashSet<short> _currentRevIndexList = new HashSet<short>(); //期望可以接收到的索引值


        //    public SlideWindowRevQueue()
        //    {
        //        _revNoOrderCache = new SortedList();

        //        _threadDataBuild = new Thread(sendNoOrderToOrder);

        //        _threadDataBuild.Start();
        //    }

        //    public RevItemHandle EnQueue(RevItemSegment item)
        //    {
        //        Monitor.Enter(obj_buildDataBagLocker);

        //        if (_currentRevIndexList.Contains(item.Index))
        //        {
        //            var tempIndex = item.Index;

        //            unchecked
        //            {
        //                tempIndex += item.Index;
        //            }

        //            _currentRevIndexList.Add(tempIndex);

        //            _revNoOrderCache.Add(item.Index, item);

        //            Monitor.PulseAll(obj_buildDataBagLocker);
        //            Monitor.Exit(obj_buildDataBagLocker);

        //            return RevItemHandle.ACK;
        //        }
        //        else
        //        {
        //            Monitor.Exit(obj_buildDataBagLocker);

        //            if (item.Index.AddAndCheckRang())
        //            {

        //            }
        //        }


        //    }

        //    //public RevItemHandle EnQueue(RevItemSegment item)
        //    //{
        //    //    RevItemHandle retValue = RevItemHandle.None;
        //    //    var temp = _lastRevBagIndex;

        //    //    if (item.Index < temp)
        //    //    {
        //    //        return RevItemHandle.None;
        //    //    }


        //    //    lock (_revNoOrderCache)
        //    //    {
        //    //        if (_revNoOrderCache.ContainsKey(item.Index))
        //    //        {
        //    //            return RevItemHandle.ACK;
        //    //        }

        //    //        //short leftCount = (short)(_revWindowsSize - _revNoOrderCache.Count);

        //    //        if (temp >= short.MaxValue - _revWindowsSize)  //发生该现象则为已经出现了数据包编号可能超过short的最大值，接收到的数据包序号可能发生回环
        //    //        {
        //    //            short tempIndex = temp;

        //    //            unchecked
        //    //            {
        //    //                tempIndex += _revWindowsSize;
        //    //            }

        //    //            if (item.Index >= temp || (item.Index > 0 && item.Index <= tempIndex))
        //    //            {
        //    //                _revNoOrderCache.Add(item.Index, item);
        //    //                retValue = RevItemHandle.ACK;
        //    //                //return true;
        //    //            }
        //    //            else
        //    //            {
        //    //                return RevItemHandle.None;
        //    //            }

        //    //        }
        //    //        else
        //    //        {
        //    //            if (item.Index >= _lastRevBagIndex && item.Index <= _lastRevBagIndex + leftCount)
        //    //            {
        //    //                _revNoOrderCache.Add(item.Index, item);
        //    //                retValue = RevItemHandle.ACK;
        //    //            }
        //    //            else
        //    //            {
        //    //                return RevItemHandle.None;
        //    //            }
        //    //        }
        //    //    }

        //    //    if (_revNoOrderCache.Count == _revWindowsSize)
        //    //    {
        //    //        unchecked
        //    //        {
        //    //            _lastRevBagIndex += _revWindowsSize;
        //    //            _lastRevBagIndex++;
        //    //        }

        //    //        RevItemSegment[] itemList = new RevItemSegment[_revWindowsSize];

        //    //        lock (_revNoOrderCache)
        //    //        {
        //    //            for (int i = 0; i < _revWindowsSize; i++)
        //    //            {
        //    //                var tempitem = _revNoOrderCache.GetByIndex(i);
        //    //                revLinkList.AddLast(tempitem);
        //    //            }

        //    //            _revNoOrderCache.Clear();
        //    //        }
        //    //    }



        //    //    //revLinkList.Add(item.Index, item);

        //    //    //for (short i = 0; i < _revNoOrderCache.Count; i++)
        //    //    //{
        //    //    //    if (_revNoOrderCache.GetByIndex(0) == _lastRevBagIndex)
        //    //    //    {
        //    //    //        revLinkList.Add(item.Index, item);
        //    //    //        _revNoOrderCache.Remove(0);

        //    //    //        unchecked
        //    //    //        {
        //    //    //            _lastRevBagIndex += 1;
        //    //    //        }
        //    //    //    }


        //    //    //    var ar = _revNoOrderCache.Values;
        //    //    //    //如果收到一个正确序号的数据片之后，就开始从无序的缓存里面查找是否有继续下去的正确序号的数据片
        //    //    //    for (short i = 0; i < ar.Count; i++)
        //    //    //    {
        //    //    //        if (_revNoOrderCache.GetByIndex(i) == _lastRevBagIndex)
        //    //    //        {
        //    //    //            revLinkList.Add(item.Index, item);
        //    //    //            _revNoOrderCache.Remove(i);

        //    //    //            unchecked
        //    //    //            {
        //    //    //                _lastRevBagIndex += 1;
        //    //    //            }
        //    //    //        }
        //    //    //    }

        //    //    lock (revLinkList)
        //    //    {
        //    //        LinkedListNode<RevItemSegment> currentNode = revLinkList.First;

        //    //        //int lastIndex = 0;

        //    //        //var lst = (IList<RevItemSegment>)revLinkList;

        //    //        //for (short i = 0; i < revLinkList.Count; i++)
        //    //        //{
        //    //        //    if (revLinkList.GetByIndex(i).IsEnd)
        //    //        //    {


        //    //        //        //TODO:申请空间，并复制从lastIndex+1 到 i 之间的所有节点的数据

        //    //        //        lastIndex = i;
        //    //        //    }
        //    //        //}



        //    //        do
        //    //        {
        //    //            if (currentNode.Value.IsEnd)
        //    //            {
        //    //                var tempList = new List<RevItemSegment>(5);

        //    //                var tempItem = revLinkList.First;

        //    //                do
        //    //                {
        //    //                    tempList.Add(tempItem.Value);
        //    //                    revLinkList.RemoveFirst();
        //    //                } while (tempItem != currentNode);

        //    //                //TODO:复制List中的数据区，并拼凑成一个数组引发收到数据的事件
        //    //            }

        //    //            currentNode = currentNode.Next;
        //    //        } while (currentNode != null);
        //    //    }

        //    //    return retValue;
        //    //}

        //    public void Close()
        //    {
        //        isClosed = true;
        //        Monitor.PulseAll(obj_buildDataBagLocker);
        //        _threadDataBuild.Join();
        //    }

        //    public event EventHandler DataBagCompleted;

        //    //private Monitor monitor = new Monitor();
        //    private object obj_buildDataBagLocker = new object();

        //    private void ContinueBuildBag()
        //    {
        //        Monitor.PulseAll(obj_buildDataBagLocker);
        //    }

        //    private void BuildDataBag(object state)
        //    {
        //        //Monitor.Enter(obj_buildDataBagLocker);
        //        var tempList = new List<RevItemSegment>(5);
        //        //do
        //        //{

        //        LinkedListNode<RevItemSegment> currentNode = revLinkList.First;

        //        do
        //        {
        //            if (currentNode.Value.IsEnd)
        //            {
        //                var tempItem = revLinkList.First;
        //                var buffer = _bufferManager.TakeBuffer(8192);
        //                var bs = new ByteStream(buffer, 0);

        //                do
        //                {
        //                    tempList.Add(tempItem.Value);

        //                    revLinkList.RemoveFirst();
        //                    tempItem.Value = null;

        //                    bs.Write(tempItem.Value.Data.Array, tempItem.Value.Data.Offset + 9, tempItem.Value.Data.Count - 9);

        //                } while (tempItem != currentNode);

        //                //TODO:复制List中的数据区，并拼凑成一个数组引发收到数据的事件

        //                OnDataBuildSuccess(new ArraySegment<byte>(buffer, 0, (int)bs.Position));

        //                tempList.Clear();
        //            }

        //            currentNode = currentNode.Next;

        //        } while (currentNode != null);


        //        //if (isClosed)
        //        //{
        //        //    break;
        //        //}

        //        //Monitor.Wait(obj_buildDataBagLocker);
        //        //} while (!isClosed);


        //        //Monitor.Exit(obj_buildDataBagLocker);
        //    }

        //    private void sendNoOrderToOrder(object state)
        //    {
        //        Monitor.Enter(obj_buildDataBagLocker);
        //        //var tempList = new List<RevItemSegment>(5);
        //        bool isNeedBuildData = false;

        //        do
        //        {

        //            if (_revNoOrderCache.Count > 0)
        //            {
        //                short lastIndex = revLinkList.Last.Value.Index;
        //                short nodeIndex = lastIndex;
        //                RevItemSegment[] tempList = new RevItemSegment[_revNoOrderCache.Count];

        //                _revNoOrderCache.CopyTo(tempList, 0);

        //                for (int i = 0; i < tempList.Length; i++)
        //                {
        //                    if (tempList[i] == null)
        //                    {
        //                        continue;
        //                    }

        //                    unchecked
        //                    {
        //                        lastIndex++;
        //                        nodeIndex += _revWindowsSize;
        //                    }

        //                    var item = (RevItemSegment)_revNoOrderCache.GetByIndex(0);
        //                    if (item.Index == lastIndex)
        //                    {
        //                        if (item.IsEnd)
        //                        {
        //                            isNeedBuildData = true;
        //                        }

        //                        _currentRevIndexList.Remove(item.Index);
        //                        _revNoOrderCache.Remove(item.Index);

        //                        _currentRevIndexList.Add(nodeIndex);

        //                        revLinkList.AddLast(item);

        //                    }
        //                    else
        //                    {
        //                        break;
        //                    }
        //                }

        //                if (isNeedBuildData)
        //                {
        //                    //Monitor.PulseAll(obj_buildDataBagLocker);
        //                    BuildDataBag(null);
        //                }

        //            }


        //            if (isClosed)
        //            {
        //                break;
        //            }


        //            Monitor.Wait(obj_buildDataBagLocker, 2000);
        //        } while (!isClosed);


        //        Monitor.Exit(obj_buildDataBagLocker);
        //    }

        //    private void OnDataBuildSuccess(ArraySegment<byte> data)
        //    {

        //    }
        //}

        //internal class SlideWindowSendQueue
        //{
        //    private ConcurrentQueue<SendItemSegment> sendQueue = new ConcurrentQueue<SendItemSegment>();
        //    private int _pageSize = 0;   //每个数据片大小
        //    private object lockerObj = new object();
        //    private short _segmentIndex = 1;

        //    public void SendEnQueue(SendItem item)
        //    {
        //        int pageSize = 0;

        //        Interlocked.Exchange(ref pageSize, _pageSize);

        //        if (item.Data.Count > pageSize)
        //        {
        //            var len = (int)Math.Ceiling((decimal)item.Data.Count / pageSize);
        //            var maxOffset = item.Data.Offset + item.Data.Count;
        //            var ar = new SendItemSegment[len];

        //            for (int i = 0; i < len; i++)
        //            {
        //                var offset = item.Data.Offset + i * pageSize;
        //                var count = Math.Min(maxOffset - offset, pageSize);
        //                var t = new ArraySegment<byte>(item.Data.Array, offset, count);

        //                ar[i] = new SendItemSegment() { Data = t, Parent = item };
        //            }

        //            ar[len - 1].IsLast = true;

        //            lock (lockerObj)
        //            {
        //                for (int i = 0; i < ar.Length; i++)
        //                {
        //                    unchecked
        //                    {
        //                        _segmentIndex++;
        //                    }

        //                    ar[i].Index = _segmentIndex;
        //                    sendQueue.Enqueue(ar[i]);
        //                }
        //            }
        //        }
        //        else
        //        {
        //            sendQueue.Enqueue(new SendItemSegment() { Data = item.Data, Index = 0, IsLast = true, Parent = item });
        //        }
        //    }

        //    public bool ACKEnQueue(ArraySegment<byte> data)
        //    {

        //    }

        //    public SendItemSegment DeQueue()
        //    {
        //        SendItemSegment page = null;

        //        if (sendQueue.TryDequeue(out page))
        //        {
        //            return page;
        //        }
        //        else
        //        {
        //            return null;
        //        }
        //    }

        //    public int Count
        //    {
        //        get { return sendQueue.Count; }
        //    }
        //}
    }


}
